package com.education.realtime.app.dwd;

import com.education.realtime.app.BaseSQLApp;
import com.education.realtime.common.Constant;
import com.education.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdInteractionFavorAdd extends BaseSQLApp {
    public static void main(String[] args) {
        new DwdInteractionFavorAdd().init(3006,"DwdInteractionFavorAddOrEducation");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"DwdInteractionFavorAddOrEducation");
        Table resultTable = tEnv.sqlQuery("select " +
                " data['id']," +
                " data['course_id']," +
                " data['user_id']," +
                " date_format(data['create_time'],'yyyy-MM-dd') date_id," +
                " data['create_time']," +
                " ts" +
                " from ods_db" +
                " where `table` = 'favor_info'" +
                " and `type` = 'insert'");
        tEnv.executeSql("create table dwd_interaction_favor_add ( " +
                "id string, " +
                "course_id string, " +
                "user_id string, " +
                "date_id string, " +
                "create_time string, " +
                "ts string " +
                ")"+ SqlUtil.getKafkaWriteDDL(Constant.TOPIC_DWD_INTERACTION_FAVOR_ADD));
        resultTable.executeInsert(Constant.TOPIC_DWD_INTERACTION_FAVOR_ADD);
    }
}
